package com.star.spring.rocketmq.annotation;

import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class OnMessageRegistrar implements BeanPostProcessor, ApplicationContextAware, DisposableBean {

    private ApplicationContext applicationContext;
    private Environment environment;
    private final List<PushConsumer> dynamicConsumers = new ArrayList<>();
    private ClientServiceProvider clientServiceProvider;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        this.environment = applicationContext.getEnvironment();
        this.clientServiceProvider = ClientServiceProvider.loadService();
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        Class<?> clazz = bean.getClass();
        Method[] methods = clazz.getDeclaredMethods();

        for (Method method : methods) {
            OnMessage onMessage = method.getAnnotation(OnMessage.class);
            if (onMessage != null) {
                createDynamicListener(bean, method, onMessage);
            }
        }

        return bean;
    }

    private void createDynamicListener(Object bean, Method method, OnMessage onMessage) {
        String topic = resolveProperty(onMessage.topic());
        String consumerGroup = resolveProperty(onMessage.consumerGroup());
        String tag = resolveProperty(onMessage.tag());
        String endpoints = resolveProperty("${rocketmq.push-consumer.endpoints}");

        try {
            // 创建客户端配置
            ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                    .setEndpoints(endpoints)
                    .build();

            // 创建过滤表达式
            FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

            // 创建推送消费者
            PushConsumer pushConsumer = clientServiceProvider.newPushConsumerBuilder()
                    .setClientConfiguration(clientConfiguration)
                    .setConsumerGroup(consumerGroup)
                    .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                    .setMessageListener(messageView -> {
                        try {
                            method.setAccessible(true);
                            Object result = method.invoke(bean, messageView);

                            if (result instanceof ConsumeResult) {
                                return (ConsumeResult) result;
                            }
                            return ConsumeResult.SUCCESS;
                        } catch (Exception e) {
                            log.error("Error consuming message for topic: {}, consumerGroup: {}", topic, consumerGroup,
                                    e);
                            return ConsumeResult.FAILURE;
                        }
                    })
                    .build();

            dynamicConsumers.add(pushConsumer);
            log.info("[OnMessage] Registered listener for topic={}, consumerGroup={}, tag={}",
                    topic, consumerGroup, tag);

        } catch (ClientException e) {
            log.error("Failed to create dynamic consumer for topic: {}, consumerGroup: {}", topic, consumerGroup, e);
        }
    }

    private String resolveProperty(String value) {
        return environment.resolvePlaceholders(value);
    }

    /**
     * 应用关闭时清理资源
     */
    @Override
    public void destroy() {
        for (PushConsumer consumer : dynamicConsumers) {
            try {
                consumer.close();
                log.info("Closed dynamic consumer successfully");
            } catch (Exception e) {
                log.error("Failed to close dynamic consumer", e);
            }
        }
        dynamicConsumers.clear();
    }
}